package com.chb.flink.sink

import java.sql.{Connection, DriverManager, PreparedStatement}

import com.chb.flink.source.{MyCustomerSource, StationLog}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

/*
 *从自定义的Source中读取StationLog数据，通过Flink写入Mysql数据库
 *
 * 当然你可以自己定义 Sink，有两种实现方式：
 * 1、实现 SinkFunction 接口。
 * 2、实现 RichSinkFunction 类。后者增加了生命周期的管理功能。
 *      比如需要在 Sink 初始化的时候创建连接对象，则最好使用第二种。
 *
 * 案例需求：把 StationLog 对象写入 Mysql 数据库中。
 */
object CustomJdbcSink {

    // 自定义Sink
    class MySqlSink extends RichSinkFunction[StationLog] {
        var conn: Connection = _
        var pst: PreparedStatement = _

        // 初始化逻辑，
        override def open(parameters: Configuration): Unit = {
            conn = DriverManager.getConnection("jdbc:mysql://localhost/test", "root", "123456")
            pst = conn.prepareStatement("insert into t_station_log(sid, call_out, call_in, call_type, call_time, duration) values(?, ?, ?, ?, ?, ?)")
        }

        // 将stationlog对象写到表中
        override def invoke(value: StationLog, context: SinkFunction.Context[_]): Unit = {
            pst.setString(1, value.sid)
            pst.setString(2, value.callOut)
            pst.setString(3, value.callIn)
            pst.setString(4, value.callType)
            pst.setLong(5, value.callTime)
            pst.setLong(6, value.duration)
            pst.executeUpdate()
        }

        override def close(): Unit = {
            pst.close()
            conn.close()
        }
    }

    def main(args: Array[String]): Unit = {
        //初始化Flink的Streaming（流计算）上下文执行环境
        val streamEnv = StreamExecutionEnvironment.getExecutionEnvironment
        streamEnv.setParallelism(1)
        //导入隐式转换
        import org.apache.flink.streaming.api.scala._

        // source
        val data: DataStream[StationLog] = streamEnv.addSource(new MyCustomerSource)

        //数据写入msyql
        data.addSink(new MySqlSink)

        streamEnv.execute()
    }
}